package com.cn.broker.processor;

import com.cn.broker.BrokerController;
import com.cn.common.ConsumeQueueData;
import com.cn.common.MessageQueue;
import com.cn.common.consumer.MessageQueueAllLock;
import com.cn.common.processor.AbstractProcessor;
import com.cn.common.processor.INettyRequestProcessor;
import com.cn.msg.Message;
import com.cn.util.RemotingSerializable;
import io.netty.channel.ChannelHandlerContext;

import java.util.Set;

/**
 * broke节点默认处理器
 */
public class BrokerAdminProcessor extends AbstractProcessor implements INettyRequestProcessor {

    private BrokerController brokerController;

    public BrokerAdminProcessor(BrokerController brokerController) {
        this.brokerController = brokerController;
    }

    @Override
    public Message process(Message message, ChannelHandlerContext ctx) {
        /**
         * 1. 拉取 offset
         * 2. 持久化offset
         * 3.
         */
        switch (message.getType()) {
            case 12:
                ConsumeQueueData consumeQueueData = RemotingSerializable.decode(message.getByteBodys(), ConsumeQueueData.class);
                this.brokerController.getConsumerOffsetManager().updateOffset(consumeQueueData);
                autoOk(message);
                break;
            case 13:
                ConsumeQueueData consumeQueueData2 = RemotingSerializable.decode(message.getByteBodys(), ConsumeQueueData.class);
                long offset = this.brokerController.getConsumerOffsetManager().readOffset(consumeQueueData2.getTopic(), (int) consumeQueueData2.getQueueId());
                consumeQueueData2.setOffset(offset);
                message.setByteBodys(RemotingSerializable.encode(consumeQueueData2));
                message.setLength(message.getByteBodys().length);
                break;
            case 15:
                //锁定queue
                MessageQueueAllLock messageQueueLock = RemotingSerializable.decode(message.getByteBodys(), MessageQueueAllLock.class);
                Set<MessageQueue> lockmq = this.brokerController.getRebalanceLockManager().lockQueue(messageQueueLock.getMessageQueueSet(), messageQueueLock.getClientId(), messageQueueLock.getGroup());
                messageQueueLock.setMessageQueueSet(lockmq);
                message.setByteBodys(RemotingSerializable.encode(messageQueueLock));
                message.setLength(message.getByteBodys().length);
//                System.out.println(lockmq);
                break;
            case 16:
                MessageQueueAllLock messageQueueAllLock = RemotingSerializable.decode(message.getByteBodys(), MessageQueueAllLock.class);
                this.brokerController.getRebalanceLockManager().unLock(messageQueueAllLock.getClientId(), messageQueueAllLock.getGroup());
                break;
        }
        return message;
    }


    public static void main(String[] args) {


    }
}
